"""
pyspark:选择，筛选与聚合

"""

from __future__ import print_function, division
from pyspark import SparkConf, SparkContext
from pyspark.sql import SparkSession

spark = SparkSession.builder.master("local").appName("test").getOrCreate()
sc = spark.sparkContext

# 读取csv文件           inferSchema自动判断数据类型默认是False
df_csv = spark.read.csv("E:\\Python\\pyspark_demo01\\pyspark_data\\ratings.csv", header=True, sep=",", inferSchema=True)

# df_csv.show()# 查看数据
# df_csv.printSchema()  # 查看数据的格式
# df_csv.describe().show()  # 查看数据基本的统计值

# 筛选栏位 userID的栏位
df_csv.select("userID").show()
# 筛选两个栏位
df_csv.select("userID", "movieID").show()

"""
使用selectExpr的时候：与操作sql差不多
"""

# 操作栏位将userID的栏位改为id
df_csv.selectExpr("userID as id").show()

# 对两个栏位进行操作
df_csv.selectExpr("movieID", "RATING * 2 as rating_2").show()

# 装换原本的数据类型
df_csv.selectExpr("cast(rating as DOUBLE)").printSchema()

"""
        筛选：filter（）
"""
# 筛选rating大于三的
df_csv.filter("rating > 3").show()
# 多条件筛选
df_csv.filter("userID == 3 and rating > 3").show()
# 多条件筛选...并按要求显示
df_csv.filter("userID == 3 and rating > 3").select("userID", "rating").show()

"""
        聚合：count()
"""

# 查看有多少数据
df_csv.count()

# 查看用过userID聚合之后的结果
df_csv.agg({"userID": "count"}).show()

# 查看给定id的和
df_csv.filter("userID = 3").count()
# 查看不同列的计算结果
df_csv.agg({"userID": "count", "rating": "avg"}).show()

# group by
# 查看每个user评论了多少部电影，评价分数如何
df_csv.groupBy("userID").agg({"movieID": "count", "rating": "avg"}).show()

# 另一种方式
# from pyspark.sql.functions import *
# df_csv.groupBy("userID").agg(count("movieID"), round(avg(df_csv.raing), 2))

